<?php


namespace EchoPHP\Rabbitmq;


class RabbitMqConnection {

    /**
     * rabbitmq 连接实例
     * @var \AMQPConnection
     */
    protected $connection;
    /**
     * 连接通道实例
     * @var \AMQPChannel
     */
    protected $channel;
    /**
     * 是否开启确认模式
     * @var bool
     */
    protected $confirmSelect = false;
    /**
     * 交换机实例
     * @var \AMQPExchange
     */
    protected $exchange;
    /**
     * 是否是默认交换机
     * @var bool
     */
    protected $defaultExchange = false;
    /**
     * 队列实例
     * @var \AMQPQueue[]
     */
    protected $queues = [];
    /**
     * @var array
     */
    protected $configs = [];

    public function __construct(\AMQPConnection $connection, array $configs) {
        $this->connection = $connection;
        $this->configs = $configs;
    }

    /**
     * 获取通信信道
     * @param string $channelName
     * @return \AMQPChannel
     * @throws \AMQPConnectionException
     */
    public function getChannel() {
        if (!$this->channel) {
            $this->declareChannel();
        }
        return $this->channel;
    }

    /**
     * 判断是否确认模式
     * @return bool
     */
    public function isConfirmSelect() {
        return $this->confirmSelect;
    }

    /**
     * 开启通道确认模式
     * @throws \AMQPConnectionException
     */
    public function channelConfirm() {
        if ($this->confirmSelect) {
            return;
        }
        $this->getChannel()->confirmSelect();
        $this->confirmSelect = true;
    }

    /**
     * 定义通道
     * @return \AMQPChannel
     * @throws \AMQPConnectionException
     */
    protected function declareChannel() {
        $this->channel = new \AMQPChannel($this->connection);
        return $this->channel;
    }

    /**
     * 获取或创建交换机
     * @return \AMQPExchange
     * @throws \AMQPChannelException
     * @throws \AMQPConnectionException
     * @throws \AMQPExchangeException
     */
    public function getExchange() {
        if (!$this->exchange) {
            $this->declareExchange();
        }
        return $this->exchange;
    }

    /**
     * 判断是否默认交换机
     * @return bool
     */
    public function isDefaultExchange() {
        return $this->defaultExchange;
    }

    /**
     * 声明交换机
     * @return \AMQPExchange
     * @throws \AMQPChannelException
     * @throws \AMQPConnectionException
     * @throws \AMQPExchangeException
     */
    protected function declareExchange() {
        $exchangeConfig = $this->configs['exchange'] ?? [];
        //创建交换机实例
        $this->exchange = new \AMQPExchange($this->getChannel());
        //匿名交换机直接返回交换机实例，即使用默认直连交换机
        if (empty($exchangeConfig['name'])) {
            $this->defaultExchange = true;
            return;
        }
        $exchangeType = $exchangeConfig['type'] ?? AMQP_EX_TYPE_DIRECT;
        if (!in_array($exchangeType, [AMQP_EX_TYPE_DIRECT, AMQP_EX_TYPE_FANOUT, AMQP_EX_TYPE_TOPIC])) {
            throw new \Exception('队列交换机类型不正确');
        }
        //定义交换机
        $this->exchange->setName($exchangeConfig['name']);
        $this->exchange->setType($exchangeType);
        //设置交换机属性
        if (isset($exchangeConfig['arguments']) && is_array($exchangeConfig['arguments'])) {
            $this->exchange->setArguments($exchangeConfig['arguments']);
        }
        //设置交换机属性
        $flags = AMQP_NOPARAM;
        //检测交换机是否存在,如果存在则返回
        if (isset($exchangeConfig['passive']) && $exchangeConfig['passive'] === true) {
            $flags += AMQP_PASSIVE;
        }
        //定义是否持久化存储
        if (isset($exchangeConfig['durable']) && $exchangeConfig['durable'] === true) {
            $flags += AMQP_DURABLE;
        }
        $this->exchange->setFlags($flags);
        $this->exchange->declareExchange();
        return $this->exchange;
    }

    /**
     * 获取或创建队列实例
     * @param string $queueName
     * @return \AMQPQueue
     * @throws \AMQPChannelException
     * @throws \AMQPConnectionException
     * @throws \AMQPExchangeException
     * @throws \AMQPQueueException
     */
    public function getQueue(string $queueName = '') {
        if (isset($this->queues[$queueName])) {
            return $this->queues[$queueName];
        }
        return $this->declareQueue($queueName);
    }

    /**
     * 声明队列
     * @param string $queueName
     * @return \AMQPQueue
     * @throws \AMQPChannelException
     * @throws \AMQPConnectionException
     * @throws \AMQPExchangeException
     * @throws \AMQPQueueException
     */
    protected function declareQueue($queueName = '') {
        //队列名称定义 [queuetype]:[else]  使用队列类型开始
        //如 notice:1 notice:2 两个队列的配置使用 notice这个类型的  parseresume 无需多个队列，直接使用类型名作为队列名即可
        $queueType = explode(':', $queueName)[0];
        $queueConfig = $this->configs['queues'][$queueType] ?? [];

        $queue = new \AMQPQueue($this->getChannel());
        //队列附加属性
        if (isset($queueConfig['arguments']) && is_array($queueConfig['arguments'])) {
            $queue->setArguments($queueConfig['arguments']);
        }
        //创建匿名队列,设置为独占队列
        $isAnonymous = (empty($queueName) || isset($queueConfig['anonymous']) && $queueConfig['anonymous'] === true) ? true : false;
        if ($isAnonymous) {
            $queue->setFlags(AMQP_EXCLUSIVE);
        } else {
            $queue->setName($queueName);
            $flags = AMQP_NOPARAM;
            if (isset($queueConfig['passive']) && $queueConfig['passive'] === true) {
                $flags += AMQP_PASSIVE;
            }
            if (isset($queueConfig['exclusive']) && $queueConfig['exclusive'] === true) {
                $flags += AMQP_EXCLUSIVE;
            }
            if (isset($queueConfig['durable']) && $queueConfig['durable'] === true) {
                $flags += AMQP_DURABLE;
            }
            if (isset($queueConfig['autodelete']) && $queueConfig['autodelete'] === true) {
                $flags += AMQP_AUTODELETE;
            }
            $queue->setFlags($flags);
        }

        $queue->declareQueue();

        //绑定交换机和队列
        $routeKeys = isset($queueConfig['routekeys']) && is_array($queueConfig['routekeys']) ? $queueConfig['routekeys'] : [];
        //默认交换机隐式绑定到所有队列，并且路由键等于队列名称，不能显示绑定到默认交换机或与之解除绑定
        if (!$this->defaultExchange) {
            //创建队列时默认建立交换机与该队列名的绑定
            $routeKeys[] = $queueName;

            $routeKeys = array_unique(array_filter($routeKeys));
            foreach ($routeKeys as $routeKey) {
                $queue->bind($this->getExchange()->getName(), $routeKey);
            }
        }

        if (!$isAnonymous) {
            $this->queues[$queueName] = $queue;
        }

        return $queue;
    }

    /**
     * 释放连接
     */
    public function disconnect() {
        $this->queues = null;
        $this->exchange = null;
        $this->defaultExchange = false;
        if (($this->channel instanceof \AMQPChannel) && $this->channel->isConnected()) {
            $this->channel->close();
        }
        $this->channel = null;
        $this->confirmSelect = false;
        if (($this->connection instanceof \AMQPConnection) && $this->connection->isConnected()) {
            if ($this->connection->isPersistent()) {
                $this->connection->pdisconnect();
            } else {
                $this->connection->disconnect();
            }
        }
        $this->connection = null;
    }
}